package com.thd.springboottest.kafka.kafkaclientedu;

        import org.apache.kafka.clients.consumer.ConsumerConfig;
        import org.apache.kafka.clients.consumer.ConsumerRecord;
        import org.apache.kafka.clients.consumer.ConsumerRecords;
        import org.apache.kafka.clients.consumer.KafkaConsumer;
        import org.junit.Test;
        import org.springframework.kafka.support.serializer.JsonDeserializer;

        import java.time.Duration;
        import java.util.Arrays;
        import java.util.Properties;

/**
 * com.thd.springboottest.kafka.kafkaclientedu.ComsumerDemo
 *
 * @author: wanglei62
 * @DATE: 2022/4/8 17:28
 **/
public class ComsumerDemo {
    @Test
    public void test01() throws InterruptedException {
        Properties props = new Properties();
        // 服务器地址和端口
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // groupID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 是否自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put("metrics.recording.level ", "DEBUG");

        // 自动提交offset间隔 , 提交offset延时(接收到消息后多久提交offset)
//        props.put("auto.commit.interval.ms", "1000");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 消费会话超时时间（超过这个时间 consumer 没有发送心跳，就会触发 rebalance 操作）
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        // 消费请求的超时时间
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "18000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("clientB: partition: %s ,topic:%s ,offset = %d, key = %s, value = %s%n",record.partition(),record.topic(), record.offset(), record.key(), record.value());
            consumer.commitSync();
        }
    }

    @Test
    public void test02() throws InterruptedException {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put("metrics.recording.level ", "DEBUG");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("clientB: partition: %s ,topic:%s ,offset = %d, key = %s, value = %s%n",record.partition(),record.topic(), record.offset(), record.key(), record.value());
            consumer.commitSync();
        }
    }

    @Test
    public void testReceiveObject() throws InterruptedException {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put("metrics.recording.level ", "DEBUG");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.springframework.kafka.support.serializer.JsonDeserializer");
        /**
         * 加到信任包，否则会报错
         * Caused by: java.lang.IllegalArgumentException: The class 'com.thd.springboottest.kafka.kafkaclientedu.JacksonBean' is not in the trusted packages: [java.util, java.lang].
         * If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).
         */
//        props.put(JsonDeserializer.TRUSTED_PACKAGES, "edu.kafka.producer.model.User, java.util, java.lang, edu.consumer.test.model, edu.consumer.test.model.*" );
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*" );


        @SuppressWarnings("resource")
        KafkaConsumer<String, JacksonBean> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("testObj"));
        while (true) {
            ConsumerRecords<String, JacksonBean> records = consumer.poll(Duration.ofMillis(5000));
            System.out.println(" ===================================================== ");
            System.out.println(" ===================================================== ");
            System.out.println(" ===================================================== ");
            for (ConsumerRecord<String, JacksonBean> record : records){
                System.out.printf("clientB: partition: %s ,offset = %d, key = %s, value = %s%n",record.partition(), record.offset(), record.key(), record.value());

            }
            consumer.commitSync();
            System.out.println(" ===================================================== ");
            System.out.println(" ===================================================== ");
            System.out.println(" ===================================================== ");
        }
    }
}
